package com.atguigu.gmall.realtime.app.dwd;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.realtime.app.func.DimSink;
import com.atguigu.gmall.realtime.app.func.TableProcessFunction;
import com.atguigu.gmall.realtime.bean.TableProcess;
import com.atguigu.gmall.realtime.utils.MyKafkaUtil;
import org.apache.flink.api.common.serialization.SerializationSchema;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer;
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaProducer;
import org.apache.flink.streaming.connectors.kafka.KafkaSerializationSchema;
import org.apache.flink.util.OutputTag;
import org.apache.kafka.clients.producer.ProducerRecord;
import javax.annotation.Nullable;


public class BaseDBApp {
    public static void main(String[] args) throws  Exception{
        //TODO 0. 基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(3);

        //设置CK参数
        env.enableCheckpointing(5000, CheckpointingMode.EXACTLY_ONCE);
        env.getCheckpointConfig().setCheckpointTimeout(60000);
//        env.setStateBackend(new FsStateBackend("hdfs://xxx:8020/gmall/flink/checkpoint"));
//        System.setProperty("HADOOP_USER_NAME","XXX");

        //TODO 1. 接收Kafka数据，过滤空值数据
        String topic = "ods_base_db_m";
        String groupId = "ods_base_group";

        //从kafka主题中读取数据
        FlinkKafkaConsumer<String> kafkaSource = MyKafkaUtil.getKafkaSource(topic,groupId);
        DataStreamSource<String> jsonDStream = env.addSource(kafkaSource);

        // 对数据进行结构的转换， String -> JSONObject
        DataStream<JSONObject> jsonStream = jsonDStream.map(
                JSON::parseObject);

        //过滤为空或者长度不足的数据
        SingleOutputStreamOperator<JSONObject> filterDStream = jsonStream.filter(
                jsonObject -> {
                    boolean flag= jsonObject.getString("table") != null
                            && jsonObject.getJSONObject("data") != null
                            && jsonObject.getString("data") != null;
                    return flag;
                }
        );

        filterDStream.print("json:::::::::");


        //TODO 5. 动态分流， 事实表到主流，协会kafka的DWD层； 维度表，通过侧输出流，写入到HBase
        OutputTag<JSONObject> hbaseTag = new OutputTag<JSONObject>(TableProcess.SINK_TYPE_HBASE){};

        SingleOutputStreamOperator<JSONObject> kafkaDStream = filterDStream.process(
                new TableProcessFunction(hbaseTag)
        );

        //获取侧输出流
        DataStream<JSONObject> hbaseDStream = kafkaDStream.getSideOutput(hbaseTag);

        kafkaDStream.print("事实表>>>>>>>>>>>>>>");
        hbaseDStream.print("维度表>>>>>>>>>>>>>>");
        hbaseDStream.addSink(new DimSink());


        FlinkKafkaProducer<JSONObject> kafkaSink = MyKafkaUtil.getKafkaSinkBySchema(
                new KafkaSerializationSchema<JSONObject>() {
                    @Override
                    public void open(SerializationSchema.InitializationContext context) throws Exception {
                        System.out.println("启动 Kafka Sink");
                    }

                    // 自定义Schema 将JSON数据转换为需要的字节格式
                    @Override
                    public ProducerRecord<byte[], byte[]> serialize(JSONObject jsonObject, @Nullable Long aLong) {
                        String  topic = jsonObject.getString("sink_table");
                        JSONObject dataJsonObj = jsonObject.getJSONObject("data");

                        return new ProducerRecord<>(topic,dataJsonObj.toJSONString().getBytes());
                    }
                }
        );

        kafkaDStream.setParallelism(3).addSink(kafkaSink);

        env.execute();
    }
}
